// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.


#include <turbo/flags/flag.h>                            // DEFINE_int32
#include <krpc/utility/compat.h>
#include <krpc/utility/fd_utility.h>                         // make_close_on_exec
#include <krpc/utility/logging.h>                            // LOG
#include <turbo/hash/m3.h> // fmix32
#include <krpc/kthread/kthread.h>                          // kthread_start_background
#include <krpc/rpc/event_dispatcher.h>
#include <krpc/rpc/config.h>

namespace krpc {

    static EventDispatcher *g_edisp = NULL;
    static pthread_once_t g_edisp_once = PTHREAD_ONCE_INIT;

    static void StopAndJoinGlobalDispatchers() {
        for (int i = 0; i < turbo::get_flag(FLAGS_task_group_ntags); ++i) {
            for (int j = 0; j < turbo::get_flag(FLAGS_event_dispatcher_num); ++j) {
                g_edisp[i * turbo::get_flag(FLAGS_event_dispatcher_num) + j].Stop();
                g_edisp[i * turbo::get_flag(FLAGS_event_dispatcher_num) + j].Join();
            }
        }
    }

    void InitializeGlobalDispatchers() {
        g_edisp = new EventDispatcher[turbo::get_flag(FLAGS_task_group_ntags) *
                                      turbo::get_flag(FLAGS_event_dispatcher_num)];
        for (int i = 0; i < turbo::get_flag(FLAGS_task_group_ntags); ++i) {
            for (int j = 0; j < turbo::get_flag(FLAGS_event_dispatcher_num); ++j) {
                kthread_attr_t attr =
                        turbo::get_flag(FLAGS_usercode_in_pthread) ? KTHREAD_ATTR_PTHREAD : KTHREAD_ATTR_NORMAL;
                attr.tag = (KTHREAD_TAG_DEFAULT + i) % turbo::get_flag(FLAGS_task_group_ntags);
                CHECK_EQ(0, g_edisp[i * turbo::get_flag(FLAGS_event_dispatcher_num) + j].Start(&attr));
            }
        }
        // This atexit is will be run before g_task_control.stop() because above
        // Start() initializes g_task_control by creating kthread (to run epoll/kqueue).
        CHECK_EQ(0, atexit(StopAndJoinGlobalDispatchers));
    }

    EventDispatcher &GetGlobalEventDispatcher(int fd, kthread_tag_t tag) {
        pthread_once(&g_edisp_once, InitializeGlobalDispatchers);
        if (turbo::get_flag(FLAGS_task_group_ntags) == 1 && turbo::get_flag(FLAGS_event_dispatcher_num) == 1) {
            return g_edisp[0];
        }
        int index = turbo::fmix32(fd) % turbo::get_flag(FLAGS_event_dispatcher_num);
        return g_edisp[tag * turbo::get_flag(FLAGS_event_dispatcher_num) + index];
    }

    int IOEventData::OnCreated(const IOEventDataOptions &options) {
        if (!options.input_cb) {
            LOG(ERROR) << "Invalid input_cb=NULL";
            return -1;
        }
        if (!options.output_cb) {
            LOG(ERROR) << "Invalid output_cb=NULL";
            return -1;
        }

        _options = options;
        return 0;
    }

    void IOEventData::BeforeRecycled() {
        _options = {NULL, NULL, NULL};
    }

} // namespace krpc

#if defined(OS_LINUX)

#include <krpc/rpc/event_dispatcher_epoll.cc>

#elif defined(OS_MACOSX)
#include <krpc/rpc/event_dispatcher_kqueue.cc>
#else
#error Not implemented
#endif
